package com.lefu.risk.storm.rule;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.utils.JedisPoolUtil;
import com.lefu.risk.storm.utils.StormCommonUtil;

public class TransTypeCountBolt extends BaseRichBolt{

	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private ConcurrentHashMap<String, Long> counts;
	private String key;
	private Logger logger;
	
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
		counts = new ConcurrentHashMap<String, Long>();
		key = StormCommonUtil.getContextDetil(context);
		logger =LoggerFactory.getLogger(TransTypeCountBolt.class);
	}

	@Override
	public void execute(Tuple input) {
		String transType =null;
		Long count = null;
		JedisPoolUtil redisUtil = null;
		try {
			redisUtil = new JedisPoolUtil(Constant.REDIS_PROPERTIES_NAME);
			transType = input.getStringByField("TRANSATION_TYPE");
			Double amount = input.getDoubleByField("AMOUNT");
			Long start = input.getLongByField("start");
			count = counts.get(transType);
			
			if(null == count) count = 0L;
			counts.put(transType, ++count);
			this.collector.emit(new Values(transType,count,start,amount));
			this.collector.ack(input);
			redisUtil.incr(key+"_ACK_"+transType);
			logger.info(key+" transType cost: "+(redisUtil.getMicrosecond()/1000 - start));
		} catch (Exception e) {
			e.printStackTrace();
			this.collector.fail(input);
			redisUtil.incr(key+"_FAIL_"+transType);
		}finally {
			if(redisUtil != null )redisUtil.close();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word","count","start","amount"));
	}

}
